package cn.itcast.czxy.BD18

import java.util.Properties

import cn.itcast.czxy.BD18.bean.Tagsfour
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object payTag {
  def main(args: Array[String]): Unit = {
    //可瑞特
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("LastTimeTag").getOrCreate()
    val url="jdbc:mysql://bd001:3306/tags_new?userUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    val tableName="tbl_basic_tag"
//谱 rua入耳 踢死~
    val properties = new Properties()
    val mysqlBD: DataFrame = spark.read.jdbc(url,tableName,properties)

    import spark.implicits._
    import org.apache.spark.sql.functions._
//入噢
    val sijibq: Map[String, String] = mysqlBD.select("rule").where("id=88").map(row => {
      row.getAs("rule").toString.split("##").map(kv => {
        val strings = kv.split("=")
        (strings(0), strings(1))
      })
    }).collectAsList().get(0).toMap

    val tagsfour: Tagsfour = getHbaseMeta(sijibq)

    val wujibq: Map[String, String] = mysqlBD.select("id", "rule").where("pid=88").map(row => {
      val id: String = row.getAs("id").toString
      val rule: String = row.getAs("rule").toString
      (rule, id)
    }).collect().toMap

    val hbaseDF: DataFrame = spark.read.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option(Tagsfour.ZKHOSTS, tagsfour.zkHosts)
      .option(Tagsfour.ZKPORT, tagsfour.zkPort)
      .option(Tagsfour.HBASETABLE, tagsfour.hbaseTable)
      .option(Tagsfour.FAMILY, tagsfour.family)
      .option(Tagsfour.SELECTFIELDS, tagsfour.selectFields)
      .load()
    val userPayTag:DataFrame = hbaseDF.groupBy('memberId, 'paymentCode).agg(count('paymentCode).as("countpay"))
//            .select('memberId,'paymentCode,'countpay,row_number().over(Window.partitionBy("memberId").orderBy(desc("countpay"))).as("rank")).show()
      .withColumn("rank", row_number().over(Window.partitionBy("memberId").orderBy(desc("countpay"))))
      .where("rank=1").select('memberId, 'paymentCode)

    val gitTagId = udf((paymentCode: String) => {
      var sj=""
      sj = wujibq.get(paymentCode).get
      if (sj==null){
        sj=wujibq.get("other").get
      }
      sj
    })
    val newUserTagid: DataFrame = userPayTag.select('memberId.as("userId"), gitTagId('paymentCode).as("tagsId"))

    val oidUserTags: DataFrame = spark.read.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option("zkHosts", tagsfour.zkHosts)
      .option("zkPort", tagsfour.zkPort)
      //存储的表
      .option("hbaseTable", "test")
      .option("family", "detail")
      .option("selectFields", "userId,tagsId")
      .load()


    val alltages= oidUserTags.join(newUserTagid,oidUserTags("userId")===newUserTagid("userId"))

    val addtag = udf((oid: String, neww: String) => {
      var tagid = ""
      if (neww == "" && oid == "") {
        tagid = ""
      } else if (oid == "") {
        tagid = neww
      } else if (neww == "") {
        tagid = oid
      } else  {
        val str: String = oid + "," + neww//1,2,3,4
        tagid= str.split(",").distinct.mkString(",")
      }
      tagid
    })

    val upTag: DataFrame = alltages.select(
      when(oidUserTags.col("userId").isNotNull, oidUserTags("userId"))
        .when(newUserTagid.col("userId").isNotNull, newUserTagid("userId"))
        .as("userId")
      ,
      addtag(oidUserTags("tagsId"), newUserTagid("tagsId")).as("tagsId")
    )

val value = upTag.repartition(4)
//        upTag.show()
    value.write.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option("zkHosts", tagsfour.zkHosts)
      .option("zkPort", tagsfour.zkPort)
      //存储的表
      .option("hbaseTable", "test")
      .option("family", "detail")
      .option("selectFields", "userId,tagsId")
      .save()

    spark.stop()


  }

  def getHbaseMeta(sijibq:Map[String,String]): Tagsfour ={
    Tagsfour(
      sijibq.getOrElse(Tagsfour.INTYPE,""),
      sijibq.getOrElse(Tagsfour.ZKHOSTS,""),
      sijibq.getOrElse(Tagsfour.ZKPORT,""),
      sijibq.getOrElse(Tagsfour.HBASETABLE,""),
      sijibq.getOrElse(Tagsfour.FAMILY,""),
      sijibq.getOrElse(Tagsfour.SELECTFIELDS,""),
      sijibq.getOrElse(Tagsfour.ROWKEY,"")
    )
  }
}
